﻿/*
* Copyright (C) 2020 ~ 2024 chaigec All Rights Reserved.
*
* Author:     chaigec <chaigec@163.com>
* Maintainer: chaigec <chaigec@163.com>

 * Use of this source code is governed by MIT license that can be found in the
 * LICENSE file in the root of the source tree. All contributing project authors
 * may be found in the AUTHORS file in the root of the source tree.
*/

#include "MkMemQueue.h"

CMkMemQueue::CMkMemQueue()
{
    m_buf = nullptr;
    m_PoolSize = 0;
    m_ProductPos = 0;
    m_ConsumerPos = 0;
    m_bRelease = FALSE;
    m_InEvent = CreateEvent(nullptr, FALSE, FALSE, nullptr);
    m_OutEvent = CreateEvent(nullptr, FALSE, FALSE, nullptr);
}

CMkMemQueue::~CMkMemQueue()
{
    Realse();
    m_PoolSize = 0;
    //MkFree(m_buf);
    if (m_buf) {
        free(m_buf);
        m_buf = nullptr;
    }
    CloseHandle(m_InEvent);
    CloseHandle(m_OutEvent);
}

CMkMemQueue *CMkMemQueue::CreateNew(Uint32 MemPoolSize)
{
    CMkMemQueue *pQueue = new CMkMemQueue;
    if (!pQueue) {
        return nullptr;
    }
    pQueue->m_buf = static_cast<Uint8 *>(malloc(MemPoolSize));
    if (!pQueue->m_buf) {
        delete pQueue;
        return nullptr;
    }
    pQueue->m_PoolSize = MemPoolSize;
    return pQueue;
}


Uint32 CMkMemQueue::Input(Uint8 *pbuf, Uint32 nLen, Uint32 WaitTime /* = 0 */)
{
    Uint32 tmpSize = 0;
    int ret = 0;
    if (!pbuf || 0 == nLen) {
        return InvalidParams;
    }
input:
    if (m_bRelease) {
        return UnknownError;
    }
    m_Lock.Lock();
    if (m_ProductPos >= m_ConsumerPos) {
        tmpSize = m_PoolSize - m_ProductPos + m_ConsumerPos - 1;
    } else {
        tmpSize = m_ConsumerPos - m_ProductPos - 1;
    }

    if (tmpSize <= nLen) {
        m_Lock.Unlock();
        //OutputDebugString("begin wait outevent\n");
        ret = WaitForSingleObject(m_OutEvent, WaitTime);
        //ResetEvent(m_OutEvent);
        //OutputDebugString("end  wait outevent\n");
        if (WAIT_OBJECT_0 == ret) {
            goto input;
        } else {
            return QueueFull;
        }
    } else {
        if (m_ProductPos + nLen < m_PoolSize) {
            memcpy(m_buf + m_ProductPos, pbuf, nLen);
            m_ProductPos += nLen;
        } else {
            memcpy(m_buf + m_ProductPos, pbuf, m_PoolSize - m_ProductPos);
            memcpy(m_buf, pbuf + m_PoolSize - m_ProductPos, m_ProductPos + nLen - m_PoolSize);
            m_ProductPos = m_ProductPos + nLen - m_PoolSize;
        }
    }
    m_Lock.Unlock();
    SetEvent(m_InEvent);
    //OutputDebugString("set input event\n");
    return NoneError;
}

Uint32 CMkMemQueue::OutPut(OUT Uint8 *pbuf, INOUT Uint32 *pLen, BOOL bRecvAll/* = TRUE*/, Uint32 WaitTime /* = 0 */)
{
    Uint32 tmpSize = 0;
    int ret = 0;
    if (!pbuf || !pLen || 0 == *pLen) {
        return InvalidParams;
    }
input:
    if (m_bRelease) {
        return UnknownError;
    }
    m_Lock.Lock();
    if (m_ConsumerPos > m_ProductPos) {
        tmpSize = m_PoolSize - m_ConsumerPos + m_ProductPos;
    } else {
        tmpSize = m_ProductPos - m_ConsumerPos;
    }

    if (tmpSize < *pLen) {
        if (0 != tmpSize && !bRecvAll) {
            if (m_ConsumerPos > m_ProductPos) {
                memcpy(pbuf, m_buf + m_ConsumerPos, m_PoolSize - m_ConsumerPos);
                memcpy(pbuf + m_PoolSize - m_ConsumerPos, m_buf, m_ProductPos);
                *pLen = m_ProductPos + m_PoolSize - m_ConsumerPos;
            } else {
                memcpy(pbuf, m_buf + m_ConsumerPos, m_ProductPos - m_ConsumerPos);
                *pLen = m_ProductPos - m_ConsumerPos;
            }
            m_ConsumerPos = m_ProductPos;
            m_Lock.Unlock();
            return NoneError;
        }
        m_Lock.Unlock();
        //char pbuf[100] = { 0 };
        //sprintf(pbuf, "begin wait input event curthread id:%x\n", GetCurrentThreadId());
        //OutputDebugString(pbuf);
        ret = WaitForSingleObject(m_InEvent, WaitTime);
        //ResetEvent(m_InEvent);
        //sprintf(pbuf, "end   wait input event curthread id:%x\n", GetCurrentThreadId());
        //OutputDebugString(pbuf);
        if (WAIT_OBJECT_0 == ret) {

            goto input;
        } else {
            return QueueEmpty;
        }
    } else {
        if (m_ConsumerPos + *pLen < m_PoolSize) {
            memcpy(pbuf, m_buf + m_ConsumerPos, *pLen);
            m_ConsumerPos += *pLen;
        } else {
            memcpy(pbuf, m_buf + m_ConsumerPos, m_PoolSize - m_ConsumerPos);
            memcpy(pbuf + m_PoolSize - m_ConsumerPos, m_buf, m_ConsumerPos + *pLen - m_PoolSize);
            m_ConsumerPos = m_ConsumerPos + *pLen - m_PoolSize;
        }
    }
    m_Lock.Unlock();
    SetEvent(m_OutEvent);
    //OutputDebugString("set ouput event\n");
    return NoneError;
}

void CMkMemQueue::Clear()
{
    m_Lock.Lock();
    m_ProductPos = m_ConsumerPos = 0;
    m_Lock.Unlock();
}

void CMkMemQueue::Realse()
{
    //OutputDebugStringA("realse\n");
    m_bRelease = TRUE;
    Clear();
    SetEvent(m_InEvent);
    SetEvent(m_OutEvent);
}


#if(!defined(WIN32) || defined(_WIN32_WCE)) && !defined(HAVE_PTHREAD_WIN32)

void *CreateEvent(char *p1, BOOL b1, BOOL b2, char *p2)
{
    MkUnUse(b1);
    MkUnUse(p1);
    MkUnUse(b2);
    MkUnUse(p2);
    EVENTHANDLE *handle = static_cast<EVENTHANDLE *>(malloc(sizeof(EVENTHANDLE)));
    pthread_condattr_t attr;
    pthread_condattr_init(&attr);
    pthread_condattr_setclock(&attr, CLOCK_MONOTONIC);
    pthread_cond_init(&handle->cond_t, nullptr/*&attr*/);
    pthread_condattr_destroy(&attr);
    pthread_mutex_init(&handle->mutex_t, nullptr);
    return handle;
}

int WaitForSingleObject(void *event_handle, unsigned long millseconds)
{
    EVENTHANDLE *handle = static_cast<EVENTHANDLE *>(event_handle);
    if (!handle) {
        return -1;
    }
    pthread_mutex_lock(&handle->mutex_t);
    struct  timespec  ts;
    clock_gettime(CLOCK_REALTIME, &ts);
    Uint64 nSec = millseconds * 1000 * 1000 + ts.tv_nsec;
    ts.tv_sec += nSec / (1000 * 1000 * 1000);
    ts.tv_nsec = nSec % (1000 * 1000 * 1000);
    int ret = pthread_cond_timedwait(&handle->cond_t, &handle->mutex_t, &ts);
    pthread_mutex_unlock(&handle->mutex_t);
    return ret;
}

void SetEvent(void *event_handle)
{
    EVENTHANDLE *handle = static_cast<EVENTHANDLE *>(event_handle);
    if (!handle) {
        return;
    }
    pthread_mutex_lock(&handle->mutex_t);
    pthread_cond_signal(&handle->cond_t);
    pthread_mutex_unlock(&handle->mutex_t);
}

void CloseHandle(void *event_handle)
{
    EVENTHANDLE *handle = static_cast<EVENTHANDLE *>(event_handle);
    if (!handle) {
        return;
    }
    pthread_cond_destroy(&handle->cond_t);
    pthread_mutex_destroy(&handle->mutex_t);
    MkFree(handle);
}
#endif
